package com.xuliugen.lsm.core;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import com.xuliugen.lsm.BloomFilter;
import com.xuliugen.lsm.util.Result;
import com.xuliugen.lsm.util.TreeUtil;
import com.xuliugen.lsm.util.Vector;
import com.xuliugen.lsm.resource.load.Content;
import com.xuliugen.lsm.resource.load.ResourceLoad;

public class Lsm<K extends Comparable<K>, V> {
    private Vector<K, V> ssTable;
    private volatile Map<K, Integer> delLog;
    private volatile Map<K, Integer> delLogBack;
    private MemTable<K, V> memTable;
    private MemTable<K, V> immutableMemTable;
    private BloomFilter<K> bloomFilter;
    private static Map<String, String> content;
    private AtomicInteger logSize;

    //默认值
    private final int default_rbTreeSize = 100;
    private final int default_delLogSize = 200;

    private final int defaultSsTableSize = 10;
    private final int default_M = 4;
    private final int default_bloomFilterSize = 5000 << 10000;

    //配置信息
    //B树阶层
    private int M;

    private int rbTreeSize;

    private int delLogSize;

    private int ssTableSize;

    private int bloomFilterSize;

    static {
        content = ResourceLoad.getContent();
    }

    Lsm() {
        init();
        this.ssTable = new Vector<>(ssTableSize);
        delLog = new ConcurrentHashMap<>();
        bloomFilter = new BloomFilter<>(bloomFilterSize);
        memTable = new MemTable<>();
        logSize = new AtomicInteger(0);
    }

    private void init() {
        String key = content.get(Content.MemtableSize);
        rbTreeSize = (key == null || key.equals("")) ? default_rbTreeSize : Integer.parseInt(key);
        key = content.get(Content.delLogSize);
        delLogSize = (key == null || key.equals("")) ? default_delLogSize : Integer.parseInt(key);
        key = content.get(Content.sstableSize);
        ssTableSize = (key == null || key.equals("")) ? defaultSsTableSize : Integer.parseInt(key);
        key = content.get(Content.bloomFilterSize);
        bloomFilterSize = (key == null || key.equals("")) ? default_bloomFilterSize : Integer.parseInt(key);
        key = content.get(Content.Bcha);
        M = (key == null || key.equals("")) ? default_M : Integer.parseInt(key);
    }

    public Result<K> put(K key, V value) {
        boolean res = memTable.add(key);
        if (res) {
            bloomFilter.add(key);
            if (memTable.size() == rbTreeSize) {
                immutableMemTable = memTable;
                memTable = new MemTable<>();
                new Thread(() -> {
                    //将红黑树转换成B树
                    ssTable.add(TreeUtil.RBTreeToBTree(immutableMemTable, M));
                    if (ssTable.size() == ssTable.getArrayLen()) {
                        new Compaction<K, V>(delLog).compaction(ssTable);
                    }
                }).start();
            }
        }
        return new Result<>(true, key);
    }

    public V get(K key) {
        if (delLog.containsKey(key)) {
            return null;
        }
        if (bloomFilter.contains(key)) {
            if (immutableMemTable != null) {
                Result<V> res1 = immutableMemTable.get(key);
                if (res1.isResult()) {
                    return res1.getMessage();
                }
            }
            return TreeUtil.findKeyFromSStable(key, ssTable);
        }
        return null;
    }

    public boolean contain(K key) {
        if (delLog.containsKey(key)) {
            return false;
        }
        if (bloomFilter.contains(key)) {
            if (immutableMemTable != null) {
                boolean res1 = immutableMemTable.get(key) != null;
                if (res1) {
                    return res1;
                }
            }
            return TreeUtil.findKeyFromSStable(key, ssTable) == null;
        }
        return false;
    }

    public boolean remove(K key) {
        //无锁操作
        while (logSize.get() == delLogSize) ;

        logSize.incrementAndGet();

        if (delLog.containsKey(key) || (delLogBack != null && delLogBack.containsKey(key))) {
            return false;
        }
        if (!memTable.remove(key).isResult()) {
            delLog.put(key, 1);
            if (delLog.size() == delLogSize) {
                delLogBack = delLog;
                delLog = new ConcurrentHashMap<>();
                //如此设计？
                logSize.addAndGet(-delLogSize);
                new Thread(() -> TreeUtil.delByDelLog(ssTable, delLogBack)).start();
            }
        } else {
            logSize.decrementAndGet();
        }
        return true;
    }

}
